package com.atguigu.gmallrealtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmallrealtime.bean.TradeSkuOrderBean;
import com.atguigu.gmallrealtime.util.HBaseUtil;
import com.atguigu.gmallrealtime.util.JedisUtil;
import com.atguigu.gmallrealtime.util.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import lombok.SneakyThrows;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author yhm
 * @create 2023-10-12 15:58
 */
public abstract class DimMapFunctionHbaseAsync<T> extends RichAsyncFunction<T, T> implements DimMapFunction<T> {

    AsyncConnection hBaseConnection;
    StatefulRedisConnection<String, String> redisConnection;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取异步连接
        hBaseConnection = HBaseUtil.getAsyncConnection();
        redisConnection = RedisUtil.getAsyncRedisConnection();
    }

    @Override
    public void close() throws Exception {
        // 关闭异步连接
//        HBaseUtil.closeAsyncConnection(hBaseConnection);
        RedisUtil.closeAsyncRedisConnection(redisConnection);
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        // 先获取redis中的数据
        try {
            CompletableFuture.<String>supplyAsync(new Supplier<String>() {

                @SneakyThrows
                @Override
                public String get() {
                    // 第一次异步访问  使用redis读取数据
                    String key = getTable()  + ":" + getId(input);
                    try {
                        return redisConnection.async().get(key).get();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                    return "";
                }
            }).thenApplyAsync(new Function<String, JSONObject>() {
                @SneakyThrows
                @Override
                public JSONObject apply(String value) {
                    // 第二次异步访问  如果redis没有数据 访问hbase
                    if (value == null || value.length() == 0) {
                        // redis当中没有数据
                        // 从HBase中读取数据
                        JSONObject dim = HBaseUtil.getDimAsync(hBaseConnection, getTable(), getId(input));
                        // 将数据写入到redis
                        String key = getTable() + ":" + getId(input);
                        redisConnection.async().setex(key, 24 * 60 * 60, dim.toJSONString());
                        return dim;
                    }
                    // 如果redis直接读取到数据  转换为jsonObject直接返回
                    return JSONObject.parseObject(value);
                }
            }).thenAccept(new Consumer<JSONObject>() {
                @SneakyThrows
                @Override
                public void accept(JSONObject dim) {
                    // 使用异步访问得到的数据  进行join合并
                    if (dim != null && dim.keySet().size() != 0) {
                        // 读取到了维度数据
                        join(input,dim);
                        resultFuture.complete(Collections.singletonList(input));
                    }
                }
            });
        } catch (Exception e) {
            System.out.println("维度关联数据错误");
            e.printStackTrace();

        }
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        // 如果获取数据超时 可以在这里报错
        System.out.println("获取数据超时" + input.toString());
    }
}
